package org.ricks.ioc;

import org.ricks.ioc.utils.AssertionUtils;
import org.ricks.ioc.utils.Logger;
import org.ricks.ioc.utils.StringUtils;
import org.ricks.ioc.utils.ThreadUtils;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static org.ricks.ioc.event.EventBus.executorOf;

/**
 * IO-intensive tasks can be executed in the Event thread pool
 * CN: Task线程池一般是用来处理客户的请求，做一些cpu密集型任务，尽量避免做一些阻塞操作；IO密集型任务可以放在Event线程池去做
 *
 * @author godotg
 */
public final class TaskBus {

    // EN: The size of the thread pool can also be specified through the provider thread configuration
    // CN: 线程池的大小，也可以通过provider thread配置指定
    public static final int EXECUTOR_SIZE;

    /**
     * EN: Use different thread pools to achieve isolation between thread pools without affecting each other
     * CN: 使用不同的线程池，让线程池之间实现隔离，互不影响
     */
    private static final ExecutorService[] executors;

    static {

        EXECUTOR_SIZE = Runtime.getRuntime().availableProcessors() + 1;

        executors = new ExecutorService[EXECUTOR_SIZE];
        for (int i = 0; i < executors.length; i++) {
            var namedThreadFactory = new TaskThreadFactory(i);
            var executor = Executors.newSingleThreadExecutor(namedThreadFactory);
            executors[i] = executor;
        }
    }

    public static class TaskThreadFactory implements ThreadFactory {
        private final int poolNumber;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final ThreadGroup group;

        public TaskThreadFactory(int poolNumber) {
            this.group = Thread.currentThread().getThreadGroup();
            this.poolNumber = poolNumber;
        }

        @Override
        public Thread newThread(Runnable runnable) {
            var threadName = StringUtils.format("task-p{}-t{}", poolNumber + 1, threadNumber.getAndIncrement());
            var thread = new Thread(group, runnable, threadName);
            thread.setDaemon(false);
            thread.setPriority(Thread.NORM_PRIORITY);
            thread.setUncaughtExceptionHandler((t, e) -> Logger.error(t.toString(), e));
            var executor = executors[poolNumber];
            AssertionUtils.notNull(executor);
            ThreadUtils.registerSingleThreadExecutor(thread, executor);
            return thread;
        }
    }


    private static int calTaskExecutorIndex(int taskExecutorHash) {
        // Other hash algorithms can be customized to make the distribution more uniform
        return Math.abs(taskExecutorHash) % EXECUTOR_SIZE;
    }

    public static int calTaskExecutorHash(Object argument) {
        var hash = 0;
        if (argument == null) {
            hash = ThreadLocalRandom.current().nextInt();
        } else if (argument instanceof Number) {
            hash = ((Number) argument).intValue();
        } else {
            hash = argument.hashCode();
        }
        return hash;
    }

    public static void execute(int taskExecutorHash, Runnable runnable) {
        Runnable safeRunnable = () -> {
            try {
                runnable.run();
            } catch (Exception e) {
                Logger.error("unknown exception", e);
            } catch (Throwable t) {
                Logger.error("unknown error", t);
            }
        };
        executors[calTaskExecutorIndex(taskExecutorHash)].execute(safeRunnable);
    }

    public static void execute(Object argument, Runnable runnable) {
        execute(calTaskExecutorHash(argument), runnable);
    }

    // 在task，event，scheduler线程执行的异步请求，请求成功过后依然在相同的线程执行回调任务
    public static Executor currentThreadExecutor() {
        var currentThreadId = Thread.currentThread().threadId();
        var executor = ThreadUtils.executorByThreadId(currentThreadId);
        return executor == null ? executorOf(ThreadLocalRandom.current().nextInt()) : executor;
    }

}